#include <iomanip>
#include <time.h>
#include <cstdio>
#ifdef POSIX
  #include <unistd.h>
#endif

#include "talk/base/helpers.h"
#include "talk/base/logging.h"
#include "talk/base/physicalsocketserver.h"
#include "talk/base/ssladapter.h"
#include "talk/xmpp/xmppclientsettings.h"
#include "talk/examples/login/xmppthread.h"
#include "talk/examples/login/xmppauth.h"
#include "talk/p2p/client/httpportallocator.h"
#include "talk/p2p/client/sessionmanagertask.h"
#include "talk/p2p/base/transport.h"
#include "presencepushtask.h"
#include "presenceouttask.h"
//#include "talk/examples/login/jingleinfotask.h"
#include "talk/session/tunnel/tunnelsessionclient.h"
#include "talk/base/host.h"
#include "talk/base/time.h"

#include "PipeClient.h"
#include "PipeEp.h"
#include "ConnectionCallback.h"
#include "XMPPCallback.h"


using namespace talk_base;

PipeEp::PipeEp(ConnectionCallback* cc, PipeClient* pump, talk_base::Thread *th, bool src, std::string& user, std::string& desc,
    std::string& max_packet_size) :
  connection_callback_(cc), pump_(pump), thread_(th), issrc_(src), user_(user), desc_(desc), message_size_str_(max_packet_size)
{
  socket_ = 0;
  stream_ = 0;
  read_buf_len_ = 0;
  write_buf_len_ = 0;
  pending_stream_read_ = 0;
  message_number_ = 0;
  stream_bytes_ = socket_bytes_ = 0;
  avg_network_time_ = 0;
  start_time_ = talk_base::Time();
  message_size_ = 0;
  bytes_at_last_print_ = 0;
  time_at_last_print_ = talk_base::Time();
  sscanf(message_size_str_.c_str(),"%d",&message_size_);
//  sscanf(message_size_str_.c_str(),"%d",message_size_);

  if(message_size_ < 64 || message_size_ > 4096)
  {
      message_size_ = 2048;
  }

  std::cout << "pipe ep object created\n";
}

PipeEp::~PipeEp(void)
{
}

void
PipeEp::StartSrc()
{
  //std::cout <<"calling tunnel to user " << user_.c_str() << "\n";
  InitiateTunnel();
  first_msg_byte_sent_at_ = 0;
  last_msg_byte_sent_at_ = 0;
}

void
PipeEp::OnMessage(Message *msg)
{
  switch (msg->message_id)
    {
  case PIPE_WRITE_SOCKET:
    //WriteToSocket();
    break;
  case PIPE_WRITE_STREAM:
    WriteToStream(0, 0);
    break;
  case PIPE_EP_START_SRC:
    StartSrc();
    break;
  case PIPE_EP_START_DST:
    printf("started dst\n");
    break;
  case PIPE_EP_SHUTDOWN:
    printf("shutting down\n");
    Shutdown();
    break;
    }
}

void
PipeEp::InitiateTunnel()
{
  std::cout <<"attempting to connect tunnel to user " << user_.c_str() << "\n";
  stream_ = NULL;
  buzz::Jid jid_arg(user_);

  char buf[10];

  // convert 123 to string [buf]
  sprintf(buf, "%d", message_size_);

  std::string message("monarch_msg_size:");
  message.append(buf);

  desc_ = message;
  //session_ = session;
  client_ = pump_->GetTunnelClient();
  //client_->session_manager().get
  stream_ = client_->CreateTunnel(jid_arg, message);

  std::cout <<"tunnel started\n";


  ProcessStream();

}

void

PipeEp::ProcessStream()
{
  stream_->SignalEvent.connect(this, &PipeEp::OnStreamEvent);
  if (stream_->GetState() == SS_OPEN)
    {
      if (issrc_)
        {
          //OnStreamEvent(stream_, SE_OPEN | SE_WRITE, 0);
        }
      else
        {
          //OnStreamEvent(stream_, SE_OPEN | SE_READ, 0);
        }
    }
}

void
PipeEp::OnStreamEvent(StreamInterface* stream, int events, int error)
{

  int err;
  if (events & SE_CLOSE)
    {
      if (error == 0)
        {
          std::cout << "Tunnel closed normally" << std::endl;
        }
      else
        {
          std::cout << "Tunnel closed with error: " << error << std::endl;
        }
      CleanupAll();
      return;
    }
  if (events & SE_OPEN)
    {
      std::cout << "Tunnel connected" << std::endl;

      connection_callback_->onTunnelConnect(this);
      if (!issrc_)
        {
          DstConnect();
        }
    }
  if ((events & SE_WRITE))
    {
      //connection_callback->onTunnelWriteable();
    }
  if ((events & SE_READ))
    {
      first_msg_byte_received_at_ = talk_base::Time();
      size_t count = 1;
      while(count > 0)
      {

        StreamResult result = stream->Read(read_buf_ + read_buf_len_, sizeof(read_buf_) - read_buf_len_, &count, &err);
        switch (result)
        {
          case SR_SUCCESS:
            // write to the socket
            //std::cout << "read "<< count <<"from stream\n";
            read_buf_len_ += count;
            connection_callback_->onDataRecive(this,read_buf_,read_buf_len_);
            read_buf_len_ = 0;
            break;
          case SR_EOS:
            printf("end of stream seen\n");
            CleanupAll();
            break;
          case SR_BLOCK:
            //printf("str blocked\n");
            break;
          default:
            printf("stream read failed\n");
            //connection_callback_->onError(this);
            ;
         }
        stream->GetAvailable(&count);
       }
    }
}

void
PipeEp::Shutdown()
{
  // cleanup the socket
  printf("cleaning up everything\n");

  if (issrc_ && socket_)
    {
      delete socket_;
    }
  // cleanup the stream
  if (stream_)
    {
      printf("cleaning up stream\n");
      stream_->Close();
      stream_ = 0;
    }

  // cleanup the PipeEp in PipeClient
  thread_->Post(pump_, PipeClient::MSG_TERMINATE_PIPE, (talk_base::MessageData*) this);
  connection_callback_->onTunnelDisconnect(this, false);
  printf("cleanup done\n");
}

void
PipeEp::CleanupAll()
{
  thread_->Post(this, PIPE_EP_SHUTDOWN);
}

//std::ostream& operator<<(std::ostream &strm, cricket::Transport &t) {
  //return  NULL; strm << t.ToString();
//}

void
PipeEp::Show(void)
{

  uint32 time_since_last_print = talk_base::TimeSince(time_at_last_print_) + 1; //in miliseconds
  uint32 bytes_since_last_print = stream_bytes_ -  bytes_at_last_print_;
  uint32 messages_since_last_print = message_number_ -  messages_at_last_print_;
  uint32 current_ = talk_base::Time();
  time_at_last_print_ = current_;


  //printf("----%s(%d,%d, %s)\n", user_.c_str(), lport_, rport_, istcp_?"tcp":"udp");
  std::cout << std::endl << "----" << user_ << " msg size = " << message_size_ << " messages sent =" << message_number_ << "----" << std::endl;


  std::cout << "total bytes sent = " << stream_bytes_ << ", ";
  std::cout << "throughput =" <<  ((bytes_since_last_print) / time_since_last_print) << " Kilobytes/s \n";
  std::cout << ((messages_since_last_print *1000) / time_since_last_print) <<" message/s \n";
  //std::cout << "avg network time =" << avg_network_time_ << " Milliseconds \n "<< std::endl;
  //std::cout << "avg_jitter_ =" << avg_jitter_ << " ms " << std::endl;
  if(session_)
  {
     // std::cout << "Type of transport =" << *(session_->GetTransport("tunnel")) << "  " << std::endl;
  }

  //std::cout << "avg write time = " << avg_write_time_ << ", ";
  //std::cout << "avg read time =" << avg_read_time_ << " ms, ";
  //std::cout << "avg number of read events per msg =" << avg_reads_per_message_ << std::endl;

  //printf("rate = %d bps, curr=%d, start=%d\n", , (int) current_, (int) start_time_);
  bytes_at_last_print_ = stream_bytes_;
  messages_at_last_print_ = message_number_;

}

void
PipeEp::ConnectTimeout(void)
{
  //if (!conn_socket_) {
  printf("connect timed out!, cleaning up\n");
  CleanupAll();
  //} else {
  //       printf("is connected!\n");
  // }
}

bool
PipeEp::DstConnect()
{

  return true;
}

void
PipeEp::OnIncomingTunnel(cricket::TunnelSessionClient* client, buzz::Jid jid, std::string description, cricket::Session* session)
{

  std::cout << "IncomingTunnel from " << jid.Str() << ": " << description << std::endl;

  session_ = session;
  client_ = client;
  description_ = description;


  message_size_ = 0;
  sscanf(description.c_str(),"monarch_msg_size:%d",&message_size_);
//  sscanf(message_size_str_.c_str(),"%d",message_size_);

  if(message_size_ < 64 || message_size_ > 4096)
  {
      message_size_ = 2048;
  }

  std::cout << "message size = " << message_size_  << "\n";


  stream_ = client_->AcceptTunnel(session_);
  ProcessStream();

}

//this returns how many bytes were sucessfully written.
int PipeEp::WriteToStream(const char* buf, size_t size)
{
  size_t count;
  int error;
  StreamResult result;
  first_msg_byte_sent_at_ = talk_base::Time();
  message_number_++;

  message_size_ = size;

  if (message_number_ % 10 == 0)
    {
      //Show();
    }

  if (!stream_)
    {
      std::cout << "Tunnel Not Open Yet" << std::endl;
      return false;
    }
  size_t write_pos = 0;
  while (write_pos < size)
    {
      result = stream_->Write(buf + write_pos, size - write_pos, &count, &error);

      if (result == SR_EOS)
        {
          std::cout << "Tunnel closed unexpectedly on write" << std::endl;
          //connection_callback_->onTunnelDisconnect(this);
          //Cleanup(stream);
          CleanupAll();
          return false;
        }

      if (result == SR_ERROR)
        {
          std::cout << "Tunnel Error" << std::endl;
          //connection_callback_->onTunnelDisconnect(this);
          //Cleanup(stream);
          CleanupAll();
          return false;
        }
      if (result == SR_SUCCESS)
        {
          write_pos += count;
          stream_bytes_ += count;
         // std::cout << "sent out " << count << "bytes" << std::endl;
        }
      if (result == SR_BLOCK)
        {
          write_pos += count;
          stream_bytes_ += count;
         // printf("write blocked write buf len=%d\n",size);
          if (size - write_pos > 0)
            {
              printf("throttling\n");
              ThreadManager::CurrentThread()->PostDelayed(50, this, PIPE_WRITE_STREAM);
            }
        }

    }

  //printf("undoing throttle\n");
  write_buf_len_ = 0;

  if (avg_write_time_ == 0)
    {
      avg_write_time_ = talk_base::TimeSince(first_msg_byte_sent_at_);
    }
  else
    {
      avg_write_time_ = (10 * avg_write_time_ + talk_base::TimeSince(first_msg_byte_sent_at_)) / 11;
    }

   return true;
}

